package com.kkb.hbase2hdfs;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

/**
 *  需求： 通过MapReduce
 *  1. 读出HBase中的myUser表的数据
 *  2. 写入到HDFS文件/user/hadoop/myuser.txt
 */
public class Main extends Configured implements Tool {
    public static void main(String[] args) throws Exception {
        Configuration configuration = HBaseConfiguration.create();
        //设置zk集群
        configuration.set("hbase.zookeeper.quorum", "master:2181,slave1:2181,slave2:2181");
        int run = ToolRunner.run(configuration, new Main(), args);
        System.exit(run);
    }

    @Override
    public int run(String[] args) throws Exception {

        //获取job对象
        Configuration conf = super.getConf();
        Job job = Job.getInstance(conf,Main.class.getSimpleName());

        //本地可以不设(设置程序运行的主类)集群运行必须
        job.setJarByClass(Main.class);

        // mapper
        TableMapReduceUtil.initTableMapperJob(TableName.valueOf("myUser"),
                new Scan(),
                HBaseReadMapper2.class,
                ImmutableBytesWritable.class,
                Result.class,
                job);
        //reducer
        //7.自定义reduce逻辑
        job.setReducerClass(HdfsWriteReducer.class);
        //k3,v3的类型
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(Text.class);

        //8.输出最终结果
        job.setOutputFormatClass(TextOutputFormat.class);
        //注意：输出路径 要不存在 否则报错
        TextOutputFormat.setOutputPath(job,new Path(args[0]));

        //提交整个job任务 等整个 任务结束之后 如果任务执行成功，返回true
        boolean b = job.waitForCompletion(true);
        //如果true 返回 0 表示整个任务执行成功
        return b?0:1;

    }
}
